package com.learning.hadoop.mapreduce.boost;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class FirstJob {

    public static void main(String[] args) {
        Configuration config = new Configuration();
        config.set("fs.defaults", "hdfs://192.168.208.106:8020");
        config.set("yarn.resourcemanager.hostname", "192.168.208.106");
//        config.set("maper.jar", "E:\\sxt\\target\\weibo1.jar");
        
        try {
            
            Job job = Job.getInstance(config);
            job.setJarByClass(FirstJob.class);
            job.setJobName("first");
            
            job.setPartitionerClass(FirstPartition.class);
            job.setMapperClass(FirstMapper.class);
            job.setNumReduceTasks(4);
            job.setCombinerClass(FirstReducer.class);
            job.setReducerClass(FirstReducer.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
            
            FileInputFormat.addInputPath(job, new Path("E:\\sxt\\1-MapReduce\\data\\weibo2.txt"));
            
            FileSystem fileSystem = FileSystem.get(config);
            
            Path outPath = new Path("E:\\sxt\\1-MapReduce\\data\\weibo1");
            if (fileSystem.exists(outPath)) {
                fileSystem.delete(outPath);
            }
            FileOutputFormat.setOutputPath(job, outPath);
            
            boolean waitForCompletion = job.waitForCompletion(true);
            if (waitForCompletion) {
                System.out.println("first success");
            }
            
        }catch (Exception e) {
            e.printStackTrace();
        }
        
    }
    
}